package com.shux.kafka.topic;

import java.util.Properties;

import org.apache.kafka.common.security.JaasUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import kafka.admin.AdminUtils;
import kafka.admin.RackAwareMode;
import kafka.server.ConfigType;
import kafka.utils.ZkUtils;

/**
 * 描述：创建、删除、查询主题相关
 * 
 * <pre>HISTORY
 * ****************************************************************************
 *  ID   DATE           PERSON          REASON
 *  1    2017年10月14日      Simba.Hua         Create
 * ****************************************************************************
 * </pre>
 * @author Simba.Hua
 */
public class KafkaTopicOperator {
	private static Logger logger = LoggerFactory.getLogger(KafkaTopicOperator.class);
	private KafkaTopicOperator(){
	}
	private static class Singleton{
		private static KafkaTopicOperator instance;
		static {
			instance = new KafkaTopicOperator();
		}
		public static KafkaTopicOperator getInstance(){
			return instance;
		}
	}
	public static KafkaTopicOperator getInstance(){
		return Singleton.getInstance();
	}
	public  boolean createTopic(String zkServer,String topicName,Integer partitions,Integer replicationFactor){
		ZkUtils zkUtils = null;
		try {
			zkUtils = ZkUtils.apply(zkServer, 30000, 30000, JaasUtils.isZkSecurityEnabled());
			AdminUtils.createTopic(zkUtils, topicName, partitions, replicationFactor, new Properties(), RackAwareMode.Enforced$.MODULE$);
			return true;
		} catch (Exception e) {
			logger.error("创建主题{}失败!",topicName,e);
		} finally {
			if(zkUtils != null)
			zkUtils.close();
		}
		return false;
	}
	
	public  boolean deleteTopic(String zkServer,String topicName){
		ZkUtils zkUtils = null;
		try {
			zkUtils = ZkUtils.apply(zkServer, 30000, 30000, JaasUtils.isZkSecurityEnabled());
			AdminUtils.deleteTopic(zkUtils, topicName);
			return true;
		} catch (Exception e) {
			logger.error("删除主题{}失败!",topicName,e);
		} finally {
			if(zkUtils != null)
			 zkUtils.close();
		}
		return false;
		
	}
	public  Properties queryTopic(String zkServer,String topicName) {
		ZkUtils zkUtils = null;
		try {
			zkUtils = ZkUtils.apply(zkServer, 30000, 30000, JaasUtils.isZkSecurityEnabled());
			Properties p = AdminUtils.fetchEntityConfig(zkUtils, ConfigType.Topic(), topicName);
			return p;
		} catch (Exception e) {
			if(zkUtils != null)
			zkUtils.close();
		}
		return null;
	}
}
